package transform;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.IntegerSerializer;

import java.util.Properties;

/**
 * @Author:RenPu
 * @Date: 2020/3/19 16:08
 * @Version: 1.0
 * @description:
 */
public class UserProducerDemo {

    public static void main(String[] args) {

        //1:创建配置对象
        Properties properties = new Properties();

        //2：配置连接kafka的集群参数
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092");

        //3：配置key的序列化器
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);

        //4：配置value的序列化器
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,UserSerializer.class);

        //5：创建生产者对象
        KafkaProducer<Integer, User> userKafkaProducer = new KafkaProducer<Integer, User>(properties);

        //6：发布消息主题
        userKafkaProducer.send(new ProducerRecord<Integer, User>("t4",1,new User("zs","男")));

        //7：提交
        userKafkaProducer.flush();

        //8：释放资源
        userKafkaProducer.close();


    }
}
